package com.idroc.cdfs.service.data;

import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.cdfs.node.INode;
import com.idroc.cdfs.node.impl.StorageNode;
import com.idroc.cdfs.service.INodeService;
import com.idroc.cdfs.util.ByteUtil;
import com.idroc.cdfs.util.ConsistentHashTable;

/**
 * 本地数据节点服务
 * @author zhandl(yuyoo4j@163.com)
 * @Aug 8, 2009 1:29:09 PM
 */
public final class LocalDataNodeService implements INodeService {
	
	private static final Logger log = LoggerFactory.getLogger(LocalDataNodeService.class);
	
	/**
	 * 本地数据节点的元数据的BerkeleyDB的目录
	 */
	public static final String META_HOME = "meta_home";
	
	public static final String STORAGE_HOME = "storage_home";
	
	public static final String REDUNDANCY_HOME = "redundancy_home";
	
	public static final String PROXY_HOME = "proxy_home";
	
	public static final String STORAGE_NODE = "storage_node";
	
	/**
	 * 文件系统的属性文件名
	 */
	public static final String CDFS_PROPERTIES = "/cdfs.properties";
	
	/**
	 * 节点所属域的类型
	 * @author zhandl 
	 *
	 */
	public static enum NodeDomain{STORAGE, PROXY, REDUNDANCY};
			 	 	 
	/**
	 * 节点服务的属性
	 */
	private Properties props = new Properties();
	
	/**
	 * 本地服务元数据保存的数据节点
	 */
	private StorageNode metaNode = new StorageNode();
	
	/**
	 * 存储节点编号
	 */
	private int storageNodeNo = -1;
	
	/**
	 * 存储节点
	 */
	private StorageNode storageNode = new StorageNode();
	 	
	/**
	 * 存储节点域
	 * 一致性哈希对象(实现consistent hash 算法)
	 */
	private ConsistentHashTable storageDomain = new ConsistentHashTable(); 
	
	/**
	 * 代理节点域
	 * 一致性哈希对象(实现consistent hash 算法)
	 */
	private ConsistentHashTable proxyDomain = new ConsistentHashTable();
	
	/**
	 * 冗余节点域
	 * 一致性哈希对象(实现consistent hash 算法)
	 */
	private ConsistentHashTable redundancyDomain = new ConsistentHashTable();

	/**
	 * 关闭服务
	 */
	public void shutdown() {

		metaNode.close();
		log.debug("关闭本地数据服务.");
	}

	/**
	 * 启动服务
	 */
	public void startup() {
		 		
		URL url = this.getClass().getResource(CDFS_PROPERTIES);		
		props = loadProps(new File(url.getPath())); 
		initMetaNode();  
		initStorageNode();
	}
	
	/**
	 * 查找特定数据项
	 * @param key -- 数据项的key
	 * @return -- 数据项的数据
	 * @memo 使用数据项的key,在整个数据服务的hashTable中hash查找节点, 并通过节点查找数据项
	 */
	public byte[] find(byte[] key) {

		int logicalNode = hashToLogicalNode(key);
		INode node = getPhysicalNode(logicalNode);
		log.debug("查找特定数据项.");
		return node.find(key);
	}
	
	/**
	 * 保存特定数据项
	 * @param key -- 数据项的key
	 * @param data -- 数据项的数据
	 * @return -- true, 当且仅当保存到berkely db
	 */
	public boolean store(final byte[] key, final byte[] data) {
		
		int logicalNode = hashToLogicalNode(key);
		INode node = getPhysicalNode(logicalNode);
		log.debug("保存特定数据项.");
		return node.store(key, data);
	}
	
	/**
	 * 删除特定数据项
	 * @param key -- 数据项的key
	 * @return true, 在Berkeley db删除了数据项,或者Berkeley db不存在此数据项
	 */
	public boolean delete(byte[] key) {
		
		int logicalNode = hashToLogicalNode(key);
		INode node = getPhysicalNode(logicalNode);
		log.debug("删除特定数据项.");
		return node.delete(key);
	}
	
	/**
	 * hash定位一致性hash环中逻辑节点
	 * @param key -- 定位节点的key
	 * @return -- 逻辑节点
	 * @memo hash定位只使用key字节数组的前两个字节
	 */
	public int hashToLogicalNode(final byte[] key) {
		
		short val = ByteUtil.getShort(key, 0);
		return Short.MAX_VALUE + val;
	}
	
	/**
	 * 获取一致性hash环中的物理节点
	 * @param logicalNode -- 逻辑节点
	 * @return -- 物理节点对象
	 */
	public INode getPhysicalNode(final int logicalNode) {
		
		INode node = (INode) storageDomain.locatePhysicalNode(logicalNode);
		log.debug("logicalNode:{}, INode:{}", logicalNode, node);
		return node;
	}
	
	/**
	 * 加载属性
	 * @param src -- 属性配置文件
	 * @return -- 属性对象
	 */
	private Properties loadProps(File src) {
		
		InputStream in = null;
		try {
			in = new FileInputStream(src);
			Properties props = new Properties();
			props.load(in);  
			return props;
		} catch (Exception ex) {
			String errorMsg = "无法加载配置文件:" + CDFS_PROPERTIES;
			log.error(errorMsg, ex);
			throw new RuntimeException(errorMsg, ex);
		} finally {
			if (null != in) {
				try { in.close(); } catch (Exception sx) { /* swallow exception */ }
			}
		}
	}
	
	/**
	 * 初始化initMetaNode属性
	 *
	 */
	private void initMetaNode() {
 
		String dbEnvPath = props.getProperty(META_HOME);
	    URL url = this.getClass().getResource(dbEnvPath.trim());
	    Properties meta = new Properties();
	    meta.put(StorageNode.NODE_STORE_HOME, url.getPath());
	    metaNode.initialize(meta);
	    metaNode.open();
		log.debug("initMetaNode 成功"); 
	}
	
	/**
	 * 初始化存储节点
	 */
	private void initStorageNode() {
		
		String nodeNo = props.getProperty(STORAGE_NODE).trim();
		String prepath = props.getProperty(STORAGE_HOME).trim(); 
	    String nodeHome = prepath + "/" + STORAGE_NODE + "_" + nodeNo;
		File dir = new File(nodeHome);
		if (!dir.exists()) {
			dir.mkdirs();
		}
		storageNodeNo = Integer.parseInt(nodeNo);
		
		Properties node = new Properties();
		node.put(StorageNode.NODE_STORE_HOME, nodeHome);
		storageNode.initialize(node);
		storageNode.open();
		log.debug("initStorageNode 成功"); 
	}
	 

	public StorageNode getMetaNode() {
		return metaNode;
	} 
	
	public int getStorageNodeNo() {
		return storageNodeNo;
	}
	
	public StorageNode getStorageNode() {
		return storageNode;
	}
	
	public Properties getProps() {
		return props;
	}
}
